package cn.xnatural.app.util;

import cn.xnatural.app.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.SocketFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocket;
import java.io.*;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.zip.GZIPInputStream;

/**
 * http 请求客户端
 */
public class Httper {
    protected static final Logger log = LoggerFactory.getLogger(Httper.class);
    protected static final Map<String, List<SocketHolder>> socketCache = new ConcurrentHashMap<>();
    protected static final long checkValidTimeout = Long.getLong("tinyHttpCheckValidTimeout", 1000L * 30);

    // 结束字符串
    protected static final String end = "\r\n";
    protected String urlStr;
    protected String contentType;
    protected String method;
    protected String bodyStr;
    protected Map<String, Param> params;
    protected Map<String, Object> cookies;
    protected final Map<String, String> requestHeader = new LinkedHashMap<>();
    public final Map<String, String> responseHeader = new LinkedHashMap<>();
    protected int connectTimeout = 3000;
    protected int readTimeout = 10000;
    protected Integer respCode;
    protected boolean debug;
    protected int ioExceptionRetryCount = 0;
    protected Charset charset = StandardCharsets.UTF_8;
    protected BiConsumer<Throwable, Httper> exHandler;
    protected Function<Map<String, String>, OutputStream> grafting;
    protected final Set<String> toStringType = new HashSet<>(Arrays.asList(
            "application/json", "text/plain", "application/javascript", "application/x-javascript", "text/html", "text/css", "text/xml"
    ));
    protected BiFunction<OutputStream, Map<String, String>, ?> resultHandler = (os, headers) -> {
        String _ct = headers.get("content-type");
        if (_ct == null || _ct.isEmpty()) {
            if (os instanceof ByteArrayOutputStream) {
                try {
                    return ((ByteArrayOutputStream) os).toString(charset.name());
                } catch (UnsupportedEncodingException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return Optional.ofNullable(_ct).map(s -> s.split(";")).map(arr -> {
            String ct = arr[0].trim().toLowerCase();
            String charset = arr.length > 1 ? arr[1].trim().split("=")[1] : this.charset.toString();
            if (toStringType.contains(ct) || urlStr.endsWith(".m3u8")) {
                if (os instanceof ByteArrayOutputStream) {
                    try {
                        return ((ByteArrayOutputStream) os).toString(charset);
                    } catch (UnsupportedEncodingException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
            return null;
        }).orElse(null);
    };
    /**
     * 记录chunked的读取状态
     */
    protected transient long chunkedLen = 0L;
    // 每秒kb限速
    protected int perSecondKb;
    protected String result = null;
    /**
     * 预请求
     */
    protected final List<Httper> preReq = new LinkedList<>();


    public Httper(String url) { this.urlStr = url; }


    public static void close() {
        for (Map.Entry<String, List<SocketHolder>> e : socketCache.entrySet()) {
            for (SocketHolder sh : e.getValue()) {
                sh.close();
            }
        }
    }


    public String get() { this.method = "GET"; return execute(); }

    /**
     * 文件处理: (文件名) -> { 返回 文件写入流}
     */
    public Httper fileHandle(Function<String, OutputStream> fileConsumer) {
        if (fileConsumer == null) return this;
        grafting = headers -> fileConsumer.apply(Optional.ofNullable(headers.get("content-disposition"))
                .map(s -> {
                    try {
                        return s.split("filename=")[1].trim();
                    } catch (Exception ignored){}
                    return null;
                }).orElse(null));
        resultHandler = (os, headers) -> {
            try {
                os.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return null;
        };
        return this;
    }

    /**
     * 处理函数: (结果流, 响应头) -> { 处理 }
     */
    public Httper resultHandle(BiFunction<OutputStream, Map<String, String>, ?> resultHandler) {
        this.resultHandler = resultHandler;
        return this;
    }
    public String post() { this.method = "POST"; return execute(); }
    public String put() { this.method = "PUT"; return execute(); }
    public String delete() { this.method = "DELETE"; return execute(); }
    public String method(String method) { this.method = method; return execute(); }
    /**
     *  设置 content-type
     * @param contentType application/json, multipart/form-data, application/x-www-form-urlencoded, text/plain
     */
    public Httper contentType(String contentType) { this.contentType = contentType; return this; }
    /**
     * response content-type 的结果需要转成 String
     */
    public Httper toStringType(String contentType) { toStringType.add(contentType); return this; }
    public Httper jsonBody(String jsonStr) {
        this.bodyStr = jsonStr;
        contentType = "application/json";
        return this;
    }
    public Httper textBody(String bodyStr) {
        this.bodyStr = bodyStr;
        contentType = "text/plain";
        return this;
    }
    public Httper formBody(String bodyStr) {
        this.bodyStr = bodyStr;
        contentType = "application/x-www-form-urlencoded";
        return this;
    }
    public Httper body(String bodyStr) {
        this.bodyStr = bodyStr;
        return this;
    }
    public Httper body(String bodyStr, String contentType) {
        this.bodyStr = bodyStr;
        this.contentType = contentType;
        return this;
    }
    public Httper readTimeout(int timeout) { this.readTimeout = timeout; return this; }
    public Httper connectTimeout(int timeout) { this.connectTimeout = timeout; return this; }
    public Httper debug() { return debug(true); }
    public Httper debug(boolean debug) { this.debug = debug; return this; }
    public Httper charset(String charset) { this.charset = Charset.forName(charset); return this; }
    public Httper id(String id) { header("x-request-id", id); return this; }
    /**
     * 添加参数
     * @param name 参数名
     * @param value 支持 {@link File}
     */
    public Httper param(String name, Object value) {
        if (params == null) params = new LinkedHashMap<>();
        params.put(name, new Param(value));
        if (value instanceof File) { contentType = "multipart/form-data"; }
        return this;
    }

    /**
     * 添加参数
     * @param name 参数名
     * @param value 支持 {@link File}
     * @param contentType 参数类型，只有是multipart/form-data才生效
     */
    public Httper param(String name, Object value, String contentType) {
        if (params == null) params = new LinkedHashMap<>();
        params.put(name, new Param(value, contentType));
        if (value instanceof File) { this.contentType = "multipart/form-data"; }
        return this;
    }

    /**
     * 添加文件流参数
     * @param name 参数名
     * @param is 文件流
     * @param length 流长度 必须 > 0
     * @param filename 文件名
     */
    public Httper fileStream(String name, InputStream is, int length, String filename) {
        if (params == null) params = new LinkedHashMap<>();
        params.put(name, new Param(new FileStream(is, length, filename)));
        contentType = "multipart/form-data";
        return this;
    }
    public Httper perSecondKb(int speed) {
        this.perSecondKb = speed;
        return this;
    }


    public Httper header(String name, String value) {
        requestHeader.put(name, value);
        return this;
    }
    public Httper digestAuth(String username, String password) {
        preReq.add(new Httper(urlStr).resultHandle((os, headers) -> {
            Object r = resultHandler.apply(os, headers);
            String hValue = headers.get("www-authenticate");
            String realm = hValue.split("realm=")[1].split(",")[0].trim().replace("\"", "");
            String qop = hValue.split("qop=")[1].split(",")[0].trim().replace("\"", "");
            String nonce = hValue.split("nonce=")[1].split(",")[0].replace("\"", "");
            String opaque = hValue.contains("opaque") ? hValue.split("opaque=")[1].split(",")[0].replace("\"", "") : null;
            String nc = "00000001";
            URI uri = URI.create(urlStr);
            String cnonce = UUID.randomUUID().toString().replace("-", "");
            StringBuilder sb = new StringBuilder();
            sb.append("Digest ").append("username=\"").append(username).append("\",realm=\"").append(realm)
                    .append("\",qop=").append(qop).append(",nonce=\"").append(nonce)
                    .append("\",uri=\"").append(uri.getPath()).append("\",nc=").append(nc)
                    .append(",cnonce=\"").append(cnonce).append("\"");
            if (opaque != null && !nonce.isEmpty()) {
                sb.append(",opaque=\"").append(opaque).append("\"");
            }
            String ha1 = Utils.md5Hex((username + ":" + realm + ":" + password).getBytes());
            String ha2 = Utils.md5Hex((method.toUpperCase() + ":" + uri.getPath()).getBytes());
            // MD5(MD5(username:realm:password):nonce:nc:cnonce:qop:MD5(<request-method>:url))
            String response = Utils.md5Hex((ha1 + ":" + nonce + ":" + nc + ":" + cnonce + ":" + qop + ":" + ha2).getBytes());
            sb.append(",response=\"").append(response).append("\"");
            header("Authorization", sb.toString());
            return r;
        }));
        return this;
    }
    public Httper basicAuth(String username, String password) {
        header("Authorization", "Basic " + new String(Base64.getEncoder().encode((username+password).getBytes())));
        return this;
    }
    public Httper cookie(String name, Object value) {
        if (cookies == null) cookies = new LinkedHashMap<>(7);
        cookies.put(name, value);
        return this;
    }
    public Httper cookies(Map<String, Object> cookies) {
        if (this.cookies == null) this.cookies = new LinkedHashMap<>(7);
        if (cookies != null) this.cookies.putAll(cookies);
        return this;
    }
    public Httper exHandler(BiConsumer<Throwable, Httper> exHandler) {
        if (this.exHandler == null) this.exHandler = exHandler;
        else {
            BiConsumer<Throwable, Httper> exist = this.exHandler;
            this.exHandler = (ex, h) -> {
                exist.accept(ex, h);
                exHandler.accept(ex, h);
            };
        }
        return this;
    }
    public Map<String, Object> cookies() {return cookies;}
    public Integer getResponseCode() {return respCode;}


    /**
     * 参数
     */
    protected static class Param {
        public final Object value;
        public final String contentType;

        public Param(Object value) {
            this(value, null);
        }

        public Param(Object value, String contentType) {
            this.value = value;
            this.contentType = contentType;
        }

        @Override
        public String toString() {
            return Objects.toString(value, null);
        }
    }


    /**
     * socket 连接持有
     */
    protected static class SocketHolder implements AutoCloseable {
        final Socket socket;
        final String key;
        final long expire = Long.getLong("tinyHttpConnectionExpire", 1000L * 60 * 30); // 连接过期
        // 上次使用时间
        long lastUsed = System.currentTimeMillis();
        final AtomicBoolean _locked = new AtomicBoolean(true);

        public SocketHolder(Socket socket, String key) {
            this.socket = socket;
            this.key = key;
        }

        @Override
        public void close() {
            if (_locked.compareAndSet(false, true)) {
                try {
                    List<SocketHolder> holders = socketCache.get(key);
                    if (holders == null) return;
                    holders.remove(this);
                    if (holders.isEmpty()) {
                        synchronized (socketCache) {
                            if (holders.isEmpty()) {
                                socketCache.remove(key);
                            }
                        }
                    }
                    socket.close();
                } catch (IOException e) {
                    log.error("close socket error: " + key, e);
                }
            }
        }

        boolean isExpired(long now) {
            return now - lastUsed > expire;
        }

        void release() { _locked.set(false); }
    }


    /**
     * 获取或创建连接
     */
    protected SocketHolder getOrCreate(URI uri) throws Exception {
        String proto = uri.toURL().getProtocol();
        boolean secure = "https".equalsIgnoreCase(proto);
        int port = uri.getPort() > 1 ? uri.getPort() : (secure ? 443 : 80);
        String key = proto + "://" + uri.getHost() + ":" + port;
        List<SocketHolder> holders = socketCache.get(key);
        if (holders == null) {
            synchronized (socketCache) {
                holders = socketCache.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>());
            }
        }
        long now = System.currentTimeMillis();
        for (Iterator<SocketHolder> it = holders.iterator(); it.hasNext(); ) {
            SocketHolder holder = it.next();
            if (holder._locked.compareAndSet(false, true)) {
                if (holder.socket.isClosed() || !holder.socket.isConnected() || !holder.socket.isBound() ||
                        holder.socket.isInputShutdown() || holder.socket.isOutputShutdown() || holder.isExpired(now)) {
                    holder.release(); holder.close();
                    continue;
                }
                // holder.socket.sendUrgentData(0xFF); // 没用(每次都创建新连接)
                if (now - holder.lastUsed > checkValidTimeout) {
                    try {
                        // 参考: org.apache.hc.core5.http.impl.io.BHttpConnectionBase#isStale
                        holder.socket.setSoTimeout(10);
                        if (holder.socket.getInputStream().read() == -1) { // 证明连接被关闭了
                            holder.release(); holder.close();
                            continue;
                        }
                    } catch (SocketTimeoutException ste) { // 证明连接可用
                    }
                }
                holder.socket.setSoTimeout(readTimeout);
                holder.lastUsed = now;

                while (it.hasNext()) { // 后面的有可能长时间不会使用，所以每次判断最后一个是否过期需要删除
                    SocketHolder h = it.next();
                    if (!it.hasNext() && h.isExpired(now)) h.close();
                }
                return holder;
            }
        }

        Socket socket = SocketFactory.getDefault().createSocket();
        socket.setKeepAlive(true);
        socket.setReuseAddress(false);
        socket.setTcpNoDelay(true);
        socket.connect(new InetSocketAddress(uri.getHost(), port), connectTimeout); // 连接
        if (secure) {
            SSLContext sc = SSLContext.getInstance("TLS");
            sc.init(null, null, null);
            socket = sc.getSocketFactory().createSocket(socket, uri.getHost(), port, true);
            SSLSocket sso = ((SSLSocket) socket);
            sso.startHandshake();
        }

        SocketHolder holder = new SocketHolder(socket, key); holders.add(holder);
        socket.setSoTimeout(readTimeout);
        log.info("New connection({})({}): {}", holders.size(), proto, socket);
        return holder;
    }


    /**
     * 执行 http 请求
     * @return http请求结果
     */
    protected String execute() {
        if (urlStr == null || urlStr.isEmpty()) throw new IllegalArgumentException("url不能为空");
        String boundary = null; // 是否为 multipart/form-data 提交
        Exception ex = null;
        SocketHolder holder = null;
        final long start = (debug || log.isDebugEnabled()) ? System.currentTimeMillis() : 0L;
        try {
            for (Httper h : preReq) {
                h.contentType(contentType).method(method);
            }
            URI uri = URI.create("GET".equalsIgnoreCase(method) ? buildUrl(urlStr, params == null ? null : params.entrySet().stream()
                    .filter(e -> e.getKey() != null && e.getValue() != null && e.getValue().value != null)
                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().value))) : urlStr);

            StringBuilder pre = new StringBuilder(); // http 首行和头部
            pre.append(method).append(" ").append(uri.getRawPath()).append(uri.getRawQuery() == null ? "" : "?" + uri.getRawQuery()).append(" HTTP/1.1");

            if (contentType == null) contentType = "application/x-www-form-urlencoded" + ";charset=" + charset;
            else if (contentType.toLowerCase().contains("multipart/form-data")) {
                // 为multipart设置boundary
                String id = requestHeader.computeIfAbsent("x-request-id", s -> UUID.randomUUID().toString().replace("-", ""));
                boundary = "boundary" + id;
                contentType = "multipart/form-data;boundary=" + boundary;
            } else if (!contentType.toLowerCase().contains(";charset=")) {
                contentType += ";charset=" + charset;
            }

            // 请求头header
            requestHeader.putIfAbsent("Accept", "*/*");
            if ("https".equals(uri.getScheme()) && uri.getPort() == 443) {
                requestHeader.putIfAbsent("Host", uri.getHost());
            } else {
                requestHeader.putIfAbsent("Host", uri.getHost() + (uri.getPort() > 0 ? ":" + uri.getPort() : ""));
            }
            requestHeader.putIfAbsent("Connection", "keep-alive");
            requestHeader.putIfAbsent("User-Agent", "Tiny");
            requestHeader.putIfAbsent("Content-Type", contentType);
            if (cookies != null) {
                requestHeader.putIfAbsent("Cookie", cookies.entrySet().stream().map(e -> e.getKey() + "=" + e.getValue()).collect(Collectors.joining(";")));
            }
            for (Map.Entry<String, String> e : requestHeader.entrySet()) {
                pre.append(end).append(e.getKey()).append(":").append(e.getValue());
            }

            holder = getOrCreate(uri);
            OutputStream os = holder.socket.getOutputStream();

            // body发送
            if (bodyStr != null && !bodyStr.isEmpty()) {
                byte[] bs = bodyStr.getBytes(charset);
                pre.append(end).append("Content-Length:").append(bs.length).append(end).append(end);
                os.write(pre.toString().getBytes(charset));
                os.write(bs);
            }
            else if (boundary != null && params != null && !params.isEmpty()) { // 多part内容
                AtomicLong contentLength = new AtomicLong();
                List<Object> piece = new LinkedList() {
                    @Override
                    public boolean add(Object o) {
                        if (o instanceof byte[]) contentLength.addAndGet(((byte[]) o).length);
                        else if (o instanceof File) contentLength.addAndGet(((File) o).length());
                        else if (o instanceof FileStream) contentLength.addAndGet(((FileStream) o).length);
                        return super.add(o);
                    }
                };

                String twoHyphens = "--";
                for (Iterator<Map.Entry<String, Param>> it = params.entrySet().stream().sorted(((o1, o2) -> {
                    // 文件发送排后边
                    if (o1.getValue().value instanceof File && !(o2.getValue().value instanceof File)) return 1;
                    else if (!(o1.getValue().value instanceof File) && o2.getValue().value instanceof File) return -1;
                    else return 0;
                })).iterator(); it.hasNext(); ) {
                    Map.Entry<String, Param> e = it.next();
                    StringBuilder sb = new StringBuilder().append(end).append(twoHyphens).append(boundary).append(end);
                    if (e.getValue().value instanceof File) {
                        piece.add(sb.append("Content-Disposition: form-data; name=\"").append(e.getKey()).append("\"; filename=\"").append(((File) e.getValue().value).getName()).append("\"").append(end)
                                .append("Content-Type: application/octet-stream").append(end).append(end)
                                .toString().getBytes(charset));
                        piece.add(e.getValue().value);
                    } else if (e.getValue().value instanceof FileStream) {
                        piece.add(sb.append("Content-Disposition: form-data; name=\"").append(e.getKey()).append("\"; filename=\"").append(((FileStream) e.getValue().value).name).append("\"").append(end)
                                .append("Content-Type: application/octet-stream").append(end).append(end)
                                .toString().getBytes(charset));
                        piece.add(e.getValue().value);
                    } else {
                        piece.add(sb.append("Content-Disposition: form-data; name=\"").append(e.getKey()).append("\"").append(end)
                                .append("Content-Type: ").append(e.getValue().contentType == null ? "text/plain" : e.getValue().contentType).append(end).append(end)
                                .append(e.getValue() == null ? "" : e.getValue().value)
                                .toString().getBytes(charset));
                    }
                }
                piece.add((end + twoHyphens + boundary + twoHyphens + end + end).getBytes(charset));

                pre.append(end).append("Content-Length:").append(contentLength.get()).append(end).append(end);
                os.write(pre.toString().getBytes(charset));
                for (Object o : piece) { // 依次写入part
                    if (o instanceof byte[]) os.write(((byte[]) o));
                    else if (o instanceof File) {
                        try (FileInputStream is = new FileInputStream((File) o)) { // copy
                            byte[] bs = new byte[5120];
                            for (int l; -1 != (l = is.read(bs)); ) {
                                os.write(bs, 0, l);
                            }
                        }
                    }
                }
            }
            else if ("POST".equalsIgnoreCase(method) && params != null && !params.isEmpty()) {
                byte[] body = params.entrySet().stream().filter(e -> e.getValue().value != null)
                        .map(e -> e.getKey() + "=" + URLEncoder.encode(e.getValue().value.toString()))
                        .collect(Collectors.joining("&"))
                        .getBytes(charset);
                pre.append(end).append("Content-Length:").append(body.length).append(end).append(end);
                os.write(pre.toString().getBytes(charset));
                os.write(body);
            }
            else {
                os.write(pre.append(end).append("Content-Length:0").append(end).append(end).toString().getBytes(charset));
            }
            os.flush();
        }
        catch (Exception e) {
            // io 异常重试
            if (e instanceof IOException && ioExceptionRetryCount++ < 1) {
                if (holder != null) {
                    holder.release(); holder.close();
                }
                return execute();
            }
            ex = e;
        }

        // 读写分开try catch
        if (ex == null && holder != null) {
            try {
                // 接收响应
                receive(holder.socket.getInputStream());
            } catch (Exception e) {
                ex = e;
            } finally {
                holder.release();
                if ("close".equalsIgnoreCase(responseHeader.get("connection"))) holder.close();
                else if (respCode == null || respCode != 200 || ex != null) holder.close();
            }
        }

        if (start > 0) {
            String id = requestHeader.get("x-request-id");
            String logMsg = "Send "+ (id == null ? "" : "(" +id+ ")") +"("+method+")" +urlStr+
                    (params == null ? "" : ", params: " + params) +
                    (bodyStr == null ? "" :", body: "+ bodyStr) +
                    (", spend: " + (System.currentTimeMillis() - start)) +
                    (", result" + (respCode == null ? "" : "(" + respCode + ")") + ": " + result);
            if (ex == null) log.info(logMsg);
            else log.error(logMsg, ex);
        }
        if (ex != null) {
            if (exHandler == null && !debug) {
                if (ex instanceof RuntimeException) throw (RuntimeException) ex;
                else throw new RuntimeException(ex);
            } else if (exHandler != null) {
                exHandler.accept(ex, this);
            }
        }
        return result;
    }


    /**
     * 接收响应数据并解析
     */
    protected void receive(InputStream is) throws Exception {
        List<ByteBuffer> datas = new LinkedList<>();
        AtomicBoolean firstLine = new AtomicBoolean(false);
        AtomicBoolean headerEnd = new AtomicBoolean(false); // 响应头读完没
        AtomicReference<OutputStream> os = new AtomicReference<>();
        AtomicReference<Long> contentLength = new AtomicReference<>();
        int logPerStep = 1024 * 1024 * 5; // 间隔多大打印一次日志
        Set<Long> logFlags = new HashSet<>();
        long contentCount = 0;
        AtomicBoolean chunked = new AtomicBoolean(false);

        long oneSecondReadCount = 0; // 统计一秒读取总数字节
        long oneSecondReadTotal = perSecondKb > 0 ? perSecondKb * 1024L : -1; // 一秒读取总数字节的限制大小
        long oneSecondStartReadTime = perSecondKb > 0 ? System.currentTimeMillis() : -1; // 一秒的开始读取时间

        int bufSize = contentLength.get() == null ? 81920 : contentLength.get() >= 81920 ? 81920 : contentLength.get().intValue();
        out: while (true) {
            byte[] bs = new byte[bufSize]; // 如果小于服务端的每次写入的大小会漏数据
            int len = is.read(bs);
            if (len == -1) throw new EOFException();
            if (len > 0) datas.add(ByteBuffer.wrap(bs, 0, len));
            if (datas.isEmpty()) continue;
            // 1. 解析首行数据
            if (!firstLine.get()) {
                if (!processLine(datas, line -> {
                    line = line.replace("\r", "");
                    String[] arr = line.split(" ");
                    respCode = Integer.valueOf(arr[1]);
                    firstLine.set(true);
                })) continue;
            }
            // 2. 解析响应头数据
            while (!headerEnd.get()) {
                boolean f = processLine(datas, line -> {
                    if ("\r".equals(line)) { // 请求头结束
                        headerEnd.set(true);
                        contentLength.set(Optional.ofNullable(responseHeader.get("content-length")).map(Long::valueOf).orElse(null));
                        chunked.set(Optional.ofNullable(responseHeader.get("transfer-encoding")).map("chunked"::equalsIgnoreCase).orElse(false));
                        grafting = grafting == null ? headers -> new ByteArrayOutputStream(
                                contentLength.get() == null ? 512 : contentLength.get().intValue()) : grafting;
                        os.set(grafting.apply(responseHeader));
                    } else {
                        int index = line.indexOf(":");
                        String hName = line.substring(0, index).toLowerCase();
                        String hValue = line.substring(index + 1).trim();
                        if ("set-cookie".equals(hName)) { // cookie 保存
                            cookie(hName, hValue.split(";")[0].split("=")[1]);
                            if (responseHeader.containsKey(hName)) {
                                hValue = responseHeader.get(hValue) + "," + hValue;
                            }
                        }
                        responseHeader.put(hName, hValue);
                    }
                });
                if (!f) continue out;
            }

            // 3. 解析内容
            if (chunked.get()) {
                contentCount += len;
                if (processChunked(datas, os.get())) break;
            } else {
                contentCount += decodeContent(datas, os.get(), contentLength.get());
                if (contentCount >= contentLength.get()) break;
            }
            long logFlag = (contentCount / logPerStep);
            if (logFlag > 0 && !logFlags.contains(logFlag)) {
                logFlags.add(logFlag);
                String id = requestHeader.get("x-request-id");
                log.info("{}Http received. {}{}", id == null || id.isEmpty() ? "" : "(" + id + ")", contentLength.get() == null ? "" : contentLength.get() + ", ", contentCount);
            }
            if (perSecondKb > 0) { // 读取速度控制
                oneSecondReadCount += len;
                if (oneSecondReadCount >= oneSecondReadTotal) {
                    long now = System.currentTimeMillis();
                    long wait = 1000 - (now - oneSecondStartReadTime);
                    if (wait > 5) { // 有等待时间, 则等下再读,
                        Thread.sleep(wait);
                        oneSecondStartReadTime = now + wait;
                    } else {
                        oneSecondStartReadTime = now;
                    }
                    oneSecondReadCount = 0;
                }
            }
        }

        // 处理结果
        result = Optional.ofNullable(resultHandler.apply(os.get(), responseHeader)).map(Object::toString).orElse(null);
    }


    /**
     * 处理行数据
     * @return true: 已处理
     */
    protected boolean processLine(List<ByteBuffer> datas, Consumer<String> lineConsumer) {
        if (datas.isEmpty()) return false;
        ByteBuffer bb1 = datas.get(0);
        if (!bb1.hasRemaining()) {
            datas.remove(0);
            return processLine(datas, lineConsumer);
        }
        String line = readLine(bb1);
        if (line != null) {
            lineConsumer.accept(line);
            return true;
        } else {
            if (datas.size() > 1) {
                ByteBuffer bb2 = datas.remove(1); // 把1位置合并到0位置的数据
                datas.set(0, ByteBuffer.allocate(bb1.remaining() + bb2.remaining()).put(bb1).put(bb2));
                return processLine(datas, lineConsumer);
            }
        }
        return false;
    }


    /**
     * 接收 chunked 数据
     * @return true: 结束
     */
    protected boolean processChunked(List<ByteBuffer> datas, OutputStream os) throws Exception {
        if (datas.isEmpty()) return false;
        if (chunkedLen < 1) {
            if (!processLine(datas, line -> {
                if ("\r".equals(line)) {
                    chunkedLen -= 1;
                } else {
                    line = line.replace("\r", "");
                    chunkedLen = Long.parseLong(line, 16);
                    if (chunkedLen == 0L) chunkedLen -= 1; // 最后结束
                }
            })) return false;
            if (chunkedLen == -2) {
                chunkedLen = 0L; // 重置
                return true;
            }
            return processChunked(datas, os);
        }
        chunkedLen -= decodeContent(datas, os, chunkedLen);
        return processChunked(datas, os);
    }


    /**
     * 内容解析
     * @param os 输出纯内容(已经处理好的)
     * @param limit 限制读取长度, null: 不限制
     * @return 长度
     */
    protected long decodeContent(List<ByteBuffer> datas, OutputStream os, Long limit) throws Exception {
        if (datas.isEmpty()) return 0L;
        String contentEncoding = responseHeader.get("content-encoding");
        long len = 0;
        for (Iterator<ByteBuffer> it = datas.iterator(); it.hasNext() && (limit == null || limit > 0); ) {
            ByteBuffer data = it.next();
            if (!data.hasRemaining()) {
                it.remove(); continue;
            }
            byte[] bs = new byte[limit == null ? data.remaining() : (int) Math.min(data.remaining(), limit)];
            len += bs.length; limit = limit == null ? null : limit - len;
            data.get(bs);
            if ("gzip".equalsIgnoreCase(contentEncoding)) {
                InputStream gis = new GZIPInputStream(new ByteArrayInputStream(bs));
                int l = 0;
                byte[] bb = new byte[512];
                while ((l = gis.read(bb)) != -1) {
                    os.write(bb, 0, l);
                }
            } else {
                os.write(bs);
            }
        }
        return len;
    }


    /**
     * 读一行数据
     * @return null: 不足一行数据
     */
    protected String readLine(ByteBuffer buf) {
        byte[] lineDelimiter = "\n".getBytes(charset);
        int index = indexOf(buf, lineDelimiter);
        if (index == -1) return null;
        int readableLength = index - buf.position();
        byte[] bs = new byte[readableLength];
        buf.get(bs);
        for (int i = 0; i < lineDelimiter.length; i++) { // 跳过 分割符的长度
            buf.get();
        }
        return new String(bs, charset);
    }


    /**
     * 查找分割符所匹配下标
     * @param buf 字节流
     * @param delim 分隔符
     * @return 下标位置
     */
    protected int indexOf(ByteBuffer buf, byte[] delim) {
        byte[] hb = buf.array();
        int delimIndex = -1; // 分割符所在的下标
        for (int i = buf.position(), size = buf.limit(); i < size; i++) {
            boolean match = true; // 是否找到和 delim 相同的字节串
            for (int j = 0; j < delim.length; j++) {
                match = match && (i + j < size) && delim[j] == hb[i + j];
            }
            if (match) {
                delimIndex = i;
                break;
            }
        }
        return delimIndex;
    }


    /**
     * 把查询参数添加到 url 后边
     * @param urlStr url
     * @param params 参数
     * @return 完整url
     */
    public static String buildUrl(String urlStr, Map<String, Object> params) {
        if (params == null || params.isEmpty()) return urlStr;
        String queryStr = params.entrySet().stream()
                .filter(e -> e.getValue() != null)
                .map(e -> e.getKey() + "=" + URLEncoder.encode(e.getValue().toString()))
                .collect(Collectors.joining("&"));
        if (urlStr.endsWith("?") || urlStr.endsWith("&")) urlStr += queryStr;
        else if (urlStr.contains("?")) urlStr += "&" + queryStr;
        else urlStr += "?" + queryStr;
        return urlStr;
    }


    protected static class FileStream {
        public final InputStream is;
        public final int length;
        // 文件名
        public String name;


        public FileStream(InputStream is, int length, String name) {
            if (is == null) throw new RuntimeException("File stream required");
            if (length < 1) throw new RuntimeException("File length incorrect");
            this.is = is;
            this.length = length;
            this.name = name == null ? "" : name.trim();
        }
    }
}
